package org.myorg.quickstart.kafka;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import java.util.Properties;

public class KafkaConsumer {

	public static void main(String[] args) throws Exception {
		final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

		Properties properties = new Properties();
		// 指定kafka server的地址，集群配多个，中间，逗号隔开
		properties.setProperty("bootstrap.servers", "slave1:9092,slave2:9092,slave3:9092");
		// 指定默认消费者group id
		properties.setProperty("group.id", "test");
		//构建FlinkKafkaConsumer
		FlinkKafkaConsumer<String> myConsumer = new FlinkKafkaConsumer<>("box-heartbeat", new SimpleStringSchema(), properties);
		//指定偏移量
		myConsumer.setStartFromEarliest();

		DataStream<String> stream = env.addSource(myConsumer);

		env.enableCheckpointing(5000);
		stream.print();

		env.execute("Flink Streaming Java API Skeleton");
	}
}
